package tcpserver

import (
	"bufio"
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"runtime"
	"sync"
	"time"
)

// 处理接口
type Handler interface {
	ServeKTV(ResponseWriter, *Request)
}

// 响应数据接口
type ResponseWriter interface {

	// the initial 512 bytes of written data to DetectContentType.
	Write([]byte) (int, error)
	WriteString(string) (int, error)
}

// A conn represents the server side of an TCP connection.
type conn struct {
	clientid int64 // 连接的编号

	localAddr  string
	remoteAddr string            // network address of remote side
	server     *Server           // the Server on which the connection arrived
	rwc        net.Conn          // i/o connection
	w          io.Writer         // checkConnErrorWriter's copy of wrc, not zeroed on Hijack
	werr       error             // any errors writing to w
	sr         liveSwitchReader  // where the LimitReader reads from; usually the rwc
	lr         *io.LimitedReader // io.LimitReader(sr)
	buf        *bufio.ReadWriter // buffered(lr,rwc), reading from bufio->limitReader->sr->rwc
	lastMethod string            // method of previous request, or ""

	token uint64 // 用户的下载令牌

	// 速度统计
	WrittenBytes int64 // 已经发送的数据量
	lastWritten  int64 // 上次的写入数据

	ReadedBytes int64 // 已经接收的字节数
	lastReaded  int64 // 上次接收到的数据

	mu           sync.Mutex // guards the following
	clientGone   bool       // if client has disconnected mid-request
	closeNotifyc chan bool  // made lazily
	hijackedv    bool       // connection has been hijacked by handler

	// 用户数据
	userdata map[string]string
}

func (c *conn) closeNotify() <-chan bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.closeNotifyc == nil {
		c.closeNotifyc = make(chan bool, 1)

		pr, pw := io.Pipe()

		readSource := c.sr.r
		c.sr.Lock()
		c.sr.r = pr
		c.sr.Unlock()
		go func() {
			_, err := io.Copy(pw, readSource)
			if err == nil {
				err = io.EOF
			}
			pw.CloseWithError(err)
			c.noteClientGone()
		}()
	}
	return c.closeNotifyc
}

func (c *conn) finalFlush() {
	if c.buf != nil {
		c.buf.Flush()

		// Steal the bufio.Reader (~4KB worth of memory) and its associated
		// reader for a future connection.
		putBufioReader(c.buf.Reader)

		// Steal the bufio.Writer (~4KB worth of memory) and its associated
		// writer for a future connection.
		putBufioWriter(c.buf.Writer)

		c.buf = nil
	}
}

// Close the connection.
func (c *conn) close() {
	c.finalFlush()
	if c.rwc != nil {
		c.rwc.Close()
		c.rwc = nil
	}

}

// rstAvoidanceDelay is the amount of time we sleep after closing the
// write side of a TCP connection before closing the entire socket.
// By sleeping, we increase the chances that the client sees our FIN
// and processes its final data before they process the subsequent RST
// from closing a connection with known unread data.
// This RST seems to occur mostly on BSD systems. (And Windows?)
// This timeout is somewhat arbitrary (~latency around the planet).
const rstAvoidanceDelay = 20 * time.Millisecond // 500 * time.Millisecond

type closeWriter interface {
	CloseWrite() error
}

var _ closeWriter = (*net.TCPConn)(nil)

// closeWrite flushes any outstanding data and sends a FIN packet (if
// client is connected via TCP), signalling that we're done.  We then
// pause for a bit, hoping the client processes it before any
// subsequent RST.
//
// See https://golang.org/issue/3595
func (c *conn) closeWriteAndWait() {
	c.finalFlush()
	if tcp, ok := c.rwc.(closeWriter); ok {
		tcp.CloseWrite()
	}
	time.Sleep(rstAvoidanceDelay)
}

func (c *conn) closeWrite() {
	c.finalFlush()
	if tcp, ok := c.rwc.(closeWriter); ok {
		tcp.CloseWrite()
	}
}

func (c *conn) noteClientGone() {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.closeNotifyc != nil && !c.clientGone {
		c.closeNotifyc <- true
	}
	c.clientGone = true
}

// 报告流量
func (c *conn) ReportTransfer() {

	// 如果TOKEN为0, 则不执行汇报
	if c.token == 0 {
		return
	}

	if c.server.SpeedReportFunc == nil {
		return
	}

	// 接收
	lastReaded := c.lastReaded
	currentReaded := c.ReadedBytes
	c.lastReaded = currentReaded

	// 发送
	lastWritten := c.lastWritten
	currentWritten := c.WrittenBytes
	c.lastWritten = currentWritten

	// 发送统计流量
	c.server.SpeedReportFunc(c.remoteAddr,
		c.token,
		currentReaded-lastReaded,   // 读取数据
		currentWritten-lastWritten) // 发送数据

}

// A switchWriter can have its Writer changed at runtime.
// It's not safe for concurrent Writes and switches.
type switchWriter struct {
	io.Writer
}

// A liveSwitchReader can have its Reader changed at runtime. It's
// safe for concurrent reads and switches, if its mutex is held.
type liveSwitchReader struct {
	sync.Mutex
	r io.Reader
	c *conn
}

func (sr *liveSwitchReader) Read(p []byte) (n int, err error) {
	sr.Lock()
	r := sr.r
	sr.Unlock()
	n, err = r.Read(p)
	// 统计发送流量
	if err == nil && n > 0 {
		sr.c.server.ReadedBytes += int64(n)
		sr.c.ReadedBytes += int64(n)
	}
	return n, err
}

// A response represents the server side of an HTTP response.
type response struct {
	conn *conn
	req  *Request // request for this response
	//wroteHeader   bool     // reply header has been (logically) written
	//wroteContinue bool     // 100 Continue response was written

	w *bufio.Writer // buffers output in chunks to chunkWriter
	//sw *switchWriter // of the bufio.Writer, for return to putBufioWriter

	WrittenBytes  int64 // number of bytes written in body
	contentLength int64 // explicitly-declared Content-Length; or -1

	// handlerHeader is the Header that Handlers get access to,
	// which may be retained and mutated even after WriteHeader.
	// handlerHeader is copied into cw.header at WriteHeader
	// time, and privately mutated thereafter.
	//handlerHeader Header
	//calledHeader bool // handler accessed handlerHeader via Header

	handlerDone bool // set true when the handler exits
}

func (w *response) Write(data []byte) (n int, err error) {
	return w.write(len(data), data, "")
}

func (w *response) WriteString(data string) (n int, err error) {
	return w.write(len(data), nil, data)
}

// either dataB or dataS is non-zero.
func (w *response) write(lenData int, dataB []byte, dataS string) (n int, err error) {

	//fmt.Printf("rsp wirte lenData=%d\n", lenData)

	if lenData == 0 {
		return 0, nil
	}

	// 速度统计
	w.WrittenBytes += int64(lenData)
	w.conn.WrittenBytes += int64(lenData)
	w.conn.server.WrittenBytes += int64(lenData) // 系统发送的数据总量

	if dataB != nil {
		return w.w.Write(dataB)
	} else {
		return w.w.WriteString(dataS)
	}
}

// 消除处理完成后执行该代码
func (w *response) finishRequest() {

	// 强制刷新输出
	w.w.Flush()

	w.handlerDone = true

}

func (w *response) CloseNotify() <-chan bool {
	return w.conn.closeNotify()
}

// 消息处理函数
type HandlerFunc func(ResponseWriter, *Request)

// ServeKTV calls f(w, r).
func (f HandlerFunc) ServeKTV(w ResponseWriter, r *Request) {
	f(w, r)
}

var errTooLarge = errors.New("http: request too large")

// Read next request from connection.
func (c *conn) readRequest() (w *response, err error) {

	// 设置读写超时时间
	if d := c.server.ReadTimeout; d != 0 {
		c.rwc.SetReadDeadline(time.Now().Add(d))
	}
	if d := c.server.WriteTimeout; d != 0 {
		defer func() {
			c.rwc.SetWriteDeadline(time.Now().Add(d))
		}()
	}

	var req *Request
	if req, err = ReadRequest(c.buf.Reader); err != nil {
		if c.lr.N == 0 {
			return nil, errTooLarge
		}
		return nil, err
	}

	req.RemoteAddr = c.remoteAddr
	req.LocalAddr = c.localAddr
	req.conn = c

	w = &response{
		conn: c,
		req:  req,
	}

	w.w = bufio.NewWriter(c.rwc)

	return w, nil
}

func (c *conn) setState(nc net.Conn, state ConnState) {
	if hook := c.server.ConnState; hook != nil {
		hook(nc, state)
	}
}

// Serve a new connection.
func (c *conn) serve() {

	// 记录TCP连接
	origConn := c.rwc // copy it before it's set nil on Close

	// 在退出当前的连接前执行以下代码
	// 防止一个TCP连接错误导致服务器崩溃
	defer func() {
		if err := recover(); err != nil {
			const size = 64 << 10
			buf := make([]byte, size)
			buf = buf[:runtime.Stack(buf, false)]
			c.server.logf("ktv: panic serving %v: %v\n%s", c.remoteAddr, err, buf)
		}

		// 关闭TCP连接
		c.close()
		c.setState(origConn, StateClosed)

		// 从用户列表删除
		if c.server != nil {
			c.server.removeConn(c)
		}

	}()

	// 连接在这里循环
	for {

		// 读取数据进行处理
		w, err := c.readRequest()
		if err != nil {
			if err == errTooLarge {

				io.WriteString(c.rwc, "HTTP/1.1 413 Request Entity Too Large\r\n\r\n")
				break
			} else if err == io.EOF {
				break // Don't reply
			} else if neterr, ok := err.(net.Error); ok && neterr.Timeout() {
				break // Don't reply
			}
			io.WriteString(c.rwc, "HTTP/1.1 400 Bad Request\r\n\r\n")
			//log.Printf("(%v) Request Error = %v", c.remoteAddr, err)
			break
		}

		// 活动
		c.setState(c.rwc, StateActive)

		// 处理消息
		serverHandler{c.server}.ServeKTV(w, w.req)

		// 消息处理完成后
		w.finishRequest()

		// 设置为空闲
		c.setState(c.rwc, StateIdle)
	} // end for

}

type SpeedReportCallbackFunc func(string, uint64, int64, int64)

// A Server defines parameters for running an HTTP server.
// The zero value for Server is a valid configuration.
type Server struct {
	Addr           string        // TCP address to listen on, ":http" if empty
	Handler        Handler       // handler to invoke, http.DefaultServeMux if nil
	ReadTimeout    time.Duration // maximum duration before timing out read of the request
	WriteTimeout   time.Duration // maximum duration before timing out write of the response
	MaxHeaderBytes int           // maximum size of request headers, DefaultMaxHeaderBytes if 0

	clientLock sync.Mutex      // guards the following
	clients    map[int64]*conn // 客户端列表

	// ConnState specifies an optional callback function that is
	// called when a client connection changes state. See the
	// ConnState type and associated constants for details.
	ConnState func(net.Conn, ConnState)

	// 地址, token, 发送字节数
	// 每一秒更新一次
	SpeedReportFunc SpeedReportCallbackFunc

	isKilled bool

	// 发送的数据量
	WrittenBytes int64 // 发送字节数
	ReadedBytes  int64 // 接收字节数

	// ErrorLog specifies an optional logger for errors accepting
	// connections and unexpected behavior from handlers.
	// If nil, logging goes to os.Stderr via the log package's
	// standard logger.
	ErrorLog *log.Logger

	disableKeepAlives int32 // accessed atomically.

	// 编号自增
	newClientid int64
}

// serverHandler delegates to either the server's Handler or
// DefaultServeMux and also handles "OPTIONS *" requests.
type serverHandler struct {
	srv *Server
}

func (sh serverHandler) ServeKTV(w ResponseWriter, req *Request) {
	handler := sh.srv.Handler
	if handler == nil {
		handler = DefaultServeMux
	}

	handler.ServeKTV(w, req)
}

// A ConnState represents the state of a client connection to a server.
// It's used by the optional Server.ConnState hook.
type ConnState int

const (
	// StateNew represents a new connection that is expected to
	// send a request immediately. Connections begin at this
	// state and then transition to either StateActive or
	// StateClosed.
	StateNew ConnState = iota

	// StateActive represents a connection that has read 1 or more
	// bytes of a request. The Server.ConnState hook for
	// StateActive fires before the request has entered a handler
	// and doesn't fire again until the request has been
	// handled. After the request is handled, the state
	// transitions to StateClosed, StateHijacked, or StateIdle.
	StateActive

	// StateIdle represents a connection that has finished
	// handling a request and is in the keep-alive state, waiting
	// for a new request. Connections transition from StateIdle
	// to either StateActive or StateClosed.
	StateIdle

	// StateClosed represents a closed connection.
	// This is a terminal state. Hijacked connections do not
	// transition to StateClosed.
	StateClosed
)

var stateName = map[ConnState]string{
	StateNew:    "new",
	StateActive: "active",
	StateIdle:   "idle",
	StateClosed: "closed",
}

func (c ConnState) String() string {
	return stateName[c]
}

// tcpKeepAliveListener sets TCP keep-alive timeouts on accepted
// connections. It's used by ListenAndServe and ListenAndServeTLS so
// dead TCP connections (e.g. closing laptop mid-download) eventually
// go away.
type tcpKeepAliveListener struct {
	*net.TCPListener
}

func (ln tcpKeepAliveListener) Accept() (c net.Conn, err error) {
	tc, err := ln.AcceptTCP()
	if err != nil {
		return
	}
	tc.SetKeepAlive(true)
	tc.SetKeepAlivePeriod(3 * time.Minute)
	return tc, nil
}

// ListenAndServe listens on the TCP network address srv.Addr and then
// calls Serve to handle requests on incoming connections.  If
// srv.Addr is blank, ":http" is used.
func (srv *Server) ListenAndServe() error {
	addr := srv.Addr
	if addr == "" {
		addr = ":http"
	}
	ln, err := net.Listen("tcp", addr)
	if err != nil {
		srv.logf("ListenAndServe error: %v", err)
		return err
	}
	err1 := srv.Serve(tcpKeepAliveListener{ln.(*net.TCPListener)})
	if err1 != nil {
		srv.logf("ListenAndServe error: %v", err1)
	}
	return err
}

// 打印信息
func (s *Server) logf(format string, args ...interface{}) {
	if s.ErrorLog != nil {
		s.ErrorLog.Printf(format, args...)
	} else {
		log.Printf(format, args...)
	}
}

// Serve accepts incoming connections on the Listener l, creating a
// new service goroutine for each.  The service goroutines read requests and
// then call srv.Handler to reply to them.
func (srv *Server) Serve(l net.Listener) error {
	defer l.Close()

	// 建立客户端列表
	if srv.clients == nil {
		srv.clients = make(map[int64]*conn, 1000)
	}

	// 启动网速统计
	if srv.SpeedReportFunc != nil {
		go srv.ClientSpeedStat()
	}

	var tempDelay time.Duration // how long to sleep on accept failure
	for {
		rw, e := l.Accept()
		if e != nil {
			if ne, ok := e.(net.Error); ok && ne.Temporary() {
				if tempDelay == 0 {
					tempDelay = 5 * time.Millisecond
				} else {
					tempDelay *= 2
				}
				if max := 1 * time.Second; tempDelay > max {
					tempDelay = max
				}
				srv.logf("http: Accept error: %v; retrying in %v", e, tempDelay)
				time.Sleep(tempDelay)
				continue
			}
			return e
		}

		go srv.HandleConnection(rw)

		tempDelay = 0
	}
}

// 处理连接
func (srv *Server) HandleConnection(rw net.Conn) {

	// 增加中间层
	nch := &ConnCommmonHandler{c: rw}
	nch.newNetConn(rw)

	if nch.isKtv != true {
		//io.WriteString(rw, "You are welcome\r\n\r\n")
		//rw.Close()
		GotoHttpServer(nch)
		return
	}

	c, err := srv.newConn(nch)
	if err != nil {
		c.close()
		return
	}
	c.setState(c.rwc, StateNew) // before Serve can return

	// 记录客户端
	srv.clientLock.Lock()
	srv.newClientid = srv.newClientid + 1
	c.clientid = srv.newClientid
	srv.clients[srv.newClientid] = c
	srv.clientLock.Unlock()

	c.serve()
}

// 客户端网速统计
func (srv *Server) ClientSpeedStat() {
	if srv.SpeedReportFunc == nil {
		return
	}

	// 如果服务器没有停止运行
	for srv.isKilled != true {

		// 每秒执行一次
		time.Sleep(time.Second * 1)

		srv.clientLock.Lock()

		for _, client := range srv.clients {

			client.ReportTransfer()

		}

		srv.clientLock.Unlock()

	}

}

// 将连接从列表中移除
func (srv *Server) removeConn(c *conn) {
	srv.clientLock.Lock()
	if srv.SpeedReportFunc != nil {
		c.ReportTransfer()
	}
	delete(srv.clients, c.clientid)
	srv.clientLock.Unlock()
}

// 停止服务
func (srv *Server) Kill() {
	srv.clientLock.Lock()
	srv.isKilled = true
	for _, client := range srv.clients {
		client.closeWrite()
	}
	srv.clientLock.Unlock()
}

// noLimit is an effective infinite upper bound for io.LimitedReader
const noLimit int64 = (1 << 63) - 1

// debugServerConnections controls whether all server connections are wrapped
// with a verbose logging wrapper.
const debugServerConnections = false

// Create new connection from rwc.
func (srv *Server) newConn(rwc net.Conn) (c *conn, err error) {
	c = new(conn)
	c.remoteAddr = rwc.RemoteAddr().String()
	c.localAddr = rwc.LocalAddr().String()
	c.server = srv
	c.rwc = rwc
	c.w = rwc
	c.userdata = make(map[string]string)

	if debugServerConnections {
		c.rwc = newLoggingConn("server", c.rwc)
	}
	c.sr.r = c.rwc
	c.sr.c = c
	c.lr = io.LimitReader(&c.sr, noLimit).(*io.LimitedReader)
	br := newBufioReader(c.lr)
	bw := newBufioWriterSize(checkConnErrorWriter{c}, 4<<10)
	c.buf = bufio.NewReadWriter(br, bw)

	return c, nil
}

var (
	bufioReaderPool   sync.Pool
	bufioWriter2kPool sync.Pool
	bufioWriter4kPool sync.Pool
)

func bufioWriterPool(size int) *sync.Pool {
	switch size {
	case 2 << 10:
		return &bufioWriter2kPool
	case 4 << 10:
		return &bufioWriter4kPool
	}
	return nil
}

func newBufioReader(r io.Reader) *bufio.Reader {
	if v := bufioReaderPool.Get(); v != nil {
		br := v.(*bufio.Reader)
		br.Reset(r)
		return br
	}
	// Note: if this reader size is every changed, update
	// TestHandlerBodyClose's assumptions.
	return bufio.NewReader(r)
}

func putBufioReader(br *bufio.Reader) {
	br.Reset(nil)
	bufioReaderPool.Put(br)
}

func newBufioWriterSize(w io.Writer, size int) *bufio.Writer {
	pool := bufioWriterPool(size)
	if pool != nil {
		if v := pool.Get(); v != nil {
			bw := v.(*bufio.Writer)
			bw.Reset(w)
			return bw
		}
	}
	return bufio.NewWriterSize(w, size)
}

func putBufioWriter(bw *bufio.Writer) {
	bw.Reset(nil)
	if pool := bufioWriterPool(bw.Available()); pool != nil {
		pool.Put(bw)
	}
}

// checkConnErrorWriter writes to c.rwc and records any write errors to c.werr.
// It only contains one field (and a pointer field at that), so it
// fits in an interface value without an extra allocation.
type checkConnErrorWriter struct {
	c *conn
}

func (w checkConnErrorWriter) Write(p []byte) (n int, err error) {
	n, err = w.c.w.Write(p) // c.w == c.rwc, except after a hijack, when rwc is nil.
	if err != nil && w.c.werr == nil {
		w.c.werr = err
	}
	return
}

// 开始服务器运行
func ListenAndServe(addr string, handler Handler) error {
	server := &Server{Addr: addr, Handler: handler}
	return server.ListenAndServe()
}

//// 协议代码处理映射
type ServeMux struct {
	mu       sync.RWMutex
	m        map[uint16]muxEntry
	NotFound muxEntry
}

//
type muxEntry struct {
	h Handler // 处理函数
	c uint16  // CMD命令码
}

// 注册处理函数
func (mux *ServeMux) Handle(cmd uint16, handler Handler) {
	mux.mu.Lock()
	defer mux.mu.Unlock()

	if cmd == 0 {
		panic("http: invalid cmd ")
	}
	if handler == nil {
		panic("http: nil handler")
	}

	mux.m[cmd] = muxEntry{h: handler, c: cmd}
}

// 注册处理函数
func (mux *ServeMux) HandleNotFound(handler Handler) {
	mux.mu.Lock()
	defer mux.mu.Unlock()

	mux.NotFound.h = handler
}

// HandleFunc registers the handler function for the given pattern.
func (mux *ServeMux) HandleFunc(cmd uint16, handler func(ResponseWriter, *Request)) {
	mux.Handle(cmd, HandlerFunc(handler))
}

// NewServeMux allocates and returns a new ServeMux.
func NewServeMux() *ServeMux { return &ServeMux{m: make(map[uint16]muxEntry)} }

// DefaultServeMux is the default ServeMux used by Serve.
var DefaultServeMux = NewServeMux()

// Handle registers the handler for the given pattern
// in the DefaultServeMux.
// The documentation for ServeMux explains how patterns are matched.
func Handle(cmd uint16, handler Handler) { DefaultServeMux.Handle(cmd, handler) }

// HandleFunc registers the handler function for the given pattern
// in the DefaultServeMux.
// The documentation for ServeMux explains how patterns are matched.
func HandleFunc(cmd uint16, handler func(ResponseWriter, *Request)) {
	DefaultServeMux.HandleFunc(cmd, handler)
}

// ServeKTV dispatches the request to the handler whose
// pattern most closely matches the request URL.
func (mux *ServeMux) ServeKTV(w ResponseWriter, r *Request) {

	for k, v := range mux.m {
		if k == r.Header.Cmd {
			h := v.h
			h.ServeKTV(w, r)
			return
		}
	}

	//  找不到对应的cmd代码时执行
	if mux.NotFound.h != nil {
		h := mux.NotFound.h
		h.ServeKTV(w, r)
	}

}

// loggingConn is used for debugging.
type loggingConn struct {
	name string
	net.Conn
}

var (
	uniqNameMu   sync.Mutex
	uniqNameNext = make(map[string]int)
)

func newLoggingConn(baseName string, c net.Conn) net.Conn {
	uniqNameMu.Lock()
	defer uniqNameMu.Unlock()
	uniqNameNext[baseName]++
	return &loggingConn{
		name: fmt.Sprintf("%s-%d", baseName, uniqNameNext[baseName]),
		Conn: c,
	}
}

func (c *loggingConn) Write(p []byte) (n int, err error) {
	log.Printf("%s.Write(%d) = ....", c.name, len(p))
	n, err = c.Conn.Write(p)
	log.Printf("%s.Write(%d) = %d, %v", c.name, len(p), n, err)
	return
}

func (c *loggingConn) Read(p []byte) (n int, err error) {
	log.Printf("%s.Read(%d) = ....", c.name, len(p))
	n, err = c.Conn.Read(p)
	log.Printf("%s.Read(%d) = %d, %v", c.name, len(p), n, err)
	return
}

func (c *loggingConn) Close() (err error) {
	log.Printf("%s.Close() = ...", c.name)
	err = c.Conn.Close()
	log.Printf("%s.Close() = %v", c.name, err)
	return
}
